package com.bj.zspace.service;

import com.bjzspace.proto.HelloProto;
import com.bjzspace.proto.HelloServiceGrpc;
import com.google.protobuf.ProtocolStringList;
import io.grpc.stub.StreamObserver;

import java.util.stream.Collectors;

public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase {

    /*
     * 1.接受client提交的参数
     * 2.业务处理 service + dao 调用对应的业务功能
     * 3.提供返回值
     */

    @Override
    public void hello(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) {
        // 1.请求参数
        String name = request.getName();
        // 2.业务处理
        System.out.println("name parameter:" + name);

        // 3.封装响应
        // 3.1 创建相应对象的构建者
        HelloProto.HelloResponse.Builder responseBuilder = HelloProto.HelloResponse.newBuilder();
        // 3.2 填充数据
        responseBuilder.setResult("hello method invoke ok");
        // 3.3 封装响应
        HelloProto.HelloResponse build = responseBuilder.build();
        responseObserver.onNext(build);
        responseObserver.onCompleted(); // 表示处理结束
    }

    /**
     * 传输多个值
     */
    @Override
    public void hello1(HelloProto.HelloRequest1 request, StreamObserver<HelloProto.HelloResponse1> responseObserver) {
        ProtocolStringList nameList = request.getNameList();
        System.out.println("method1 invoke ok");
        for (int i = 0; i < nameList.size(); i++) {
            System.out.println("hello1 method name["+i+"]=" + request.getName(i));
        }
        HelloProto.HelloResponse1.Builder builder = HelloProto.HelloResponse1.newBuilder();
        builder.setResult("result=" + nameList.stream().collect(Collectors.joining(",")));
        HelloProto.HelloResponse1 helloResponse1 = builder.build();
        responseObserver.onNext(helloResponse1);
        responseObserver.onCompleted();

    }

    /**
     * 服务端流式RPC
     */
    @Override
    public void c2ss(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) {

        System.out.println("c2ss invoke ok");
        String name = request.getName();
        System.out.println("request name:" + name);

        for (int i = 0; i < 9; i++) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
            builder.setResult("hello name:" + name + i);

            HelloProto.HelloResponse helloResponse = builder.build();
            responseObserver.onNext(helloResponse);
        }
        responseObserver.onCompleted();

    }

    /**
     * 客户端流式RPC
     */
    @Override
    public StreamObserver<HelloProto.HelloRequest> cs2s(StreamObserver<HelloProto.HelloResponse> responseObserver) {
        return new StreamObserver<HelloProto.HelloRequest>() {
            @Override
            public void onNext(HelloProto.HelloRequest value) {
                System.out.println("cs2s result:" + value.getName());
            }

            @Override
            public void onError(Throwable t) {
                System.out.println("cs2s error:" + t.getMessage());
            }
            @Override
            public void onCompleted() {
                System.out.println("cs2s onCompleted");
                HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
                builder.setResult("cs2s invoke result");
                HelloProto.HelloResponse response = builder.build();
                responseObserver.onNext(response);
                responseObserver.onCompleted();
            }
        };
    }

    /**
     * 双向流式RPC
     * @param responseObserver
     * @return
     */
    @Override
    public StreamObserver<HelloProto.HelloRequest> cs2ss(StreamObserver<HelloProto.HelloResponse> responseObserver) {



        return new StreamObserver<HelloProto.HelloRequest>() {
            @Override
            public void onNext(HelloProto.HelloRequest value) {
                System.out.println("cs2ss onNext:" + value.getName());
                HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
                builder.setResult("ack " + value.getName());
                responseObserver.onNext(builder.build());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void onError(Throwable t) {
                System.out.println("cs2ss onError:" + t.getMessage());
            }
            @Override
            public void onCompleted() {
                System.out.println("cs2ss onCompleted");
                responseObserver.onCompleted();
            }
        };

    }
}
